package com.neusoft.databus.core.service.impl;

import java.io.IOException;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;

import javax.servlet.http.HttpServletResponse;
import javax.transaction.Transactional;

import org.apache.commons.lang3.StringUtils;
import org.apache.poi.xssf.streaming.SXSSFWorkbook;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.querydsl.QPageRequest;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import com.excel.poi.ExcelBoot;
import com.excel.poi.entity.ErrorEntity;
import com.excel.poi.function.ExportFunction;
import com.excel.poi.function.ImportFunction;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.neusoft.bizcore.common.utils.DateUtil;
import com.neusoft.bizcore.web.exception.CustomRuntimeException;
import com.neusoft.bizcore.web.support.Searchable;
import com.neusoft.bizcore.web.utils.ExcelPoiBoot;
import com.neusoft.bizcore.web.utils.ExcelUtil;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.common.constant.DatabusErrorCode;
import com.neusoft.databus.common.dto.GatewayDTO;
import com.neusoft.databus.common.dto.ResourceDTO;
import com.neusoft.databus.common.enums.PhysicalConnType;
import com.neusoft.databus.common.enums.ResourceWorkMode;
import com.neusoft.databus.common.model.Gateway;
import com.neusoft.databus.core.constant.DatabusCoreConstant;
import com.neusoft.databus.core.converter.GatewayConverter;
import com.neusoft.databus.core.event.EventType;
import com.neusoft.databus.core.event.EventWrapper;
import com.neusoft.databus.core.event.GatewayBatchChangedEvent;
import com.neusoft.databus.core.event.GatewayDeleteEvent;
import com.neusoft.databus.core.event.GatewayStatusChangedEvent;
import com.neusoft.databus.core.repository.GatewayRepository;
import com.neusoft.databus.core.service.GatewayService;
import com.neusoft.databus.core.service.ResourceService;
import com.neusoft.databus.core.vo.GatewayPortVO;

import lombok.extern.slf4j.Slf4j;

/**
 * 网关服务
 *
 * @author sunchf
 * @date 2019年1月17日 上午10:35:31
 */
@Slf4j
@Service
@Transactional
public class GatewayServiceImpl implements GatewayService {

    @Autowired
    private GatewayConverter gatewayConverter;
    @Autowired
    private GatewayRepository gatewayRepository;
    @Autowired
    private ResourceService resourceService;
    @Autowired
    StringRedisTemplate template;

    @Value("${databus.core.redis.subscribeChannel: channel:databus}")
    private String redisSubscribeChannel;

    //    @javax.annotation.Resource(name = "modelUpdateEventBus")
    //    private AsyncEventBus modelUpdateEventBus;

    //    @PostConstruct
    //    public void init() {
    //        this.modelUpdateEventBus.register(this);
    //    }

    @Override
    public GatewayDTO get(final long id) {
        final Gateway model = this.gatewayRepository.getOne(id);
        if (null == model) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_404001, new Object[] {id });
        }
        return this.gatewayConverter.toDTO(model);
    }

    @Override
    public GatewayDTO findByCode(final String code) {
        final Gateway model = this.gatewayRepository.findByCode(code);
        if (null == model) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_404001, new Object[] {code });
        }
        return this.gatewayConverter.toDTO(model);
    }

    @Override
    public GatewayDTO findByCodeWithResources(final String code) {
        final Gateway model = this.gatewayRepository.findByCode(code);
        if (null == model) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_404001, new Object[] {code });
        }
        final GatewayDTO dto = this.gatewayConverter.toDTO(model);
        final List<ResourceDTO> resources = this.resourceService.findByGateway(code);
        dto.getResources().addAll(resources);
        return dto;
    }

    @Override
    public List<GatewayDTO> search(final Searchable searchable) {
        final List<Gateway> models = this.gatewayRepository.search(searchable);
        return this.gatewayConverter.toListDTO(models);
    }

    @Override
    public Page<GatewayDTO> search(final Pageable pageable, final Searchable searchable) {
        final Page<Gateway> models = this.gatewayRepository.search(pageable, searchable);
        return this.gatewayConverter.toPageDTO(models);
    }

    @Override
    public GatewayDTO create(final GatewayDTO dto) {
        final Gateway model = this.gatewayConverter.toModel(dto);
        this.gatewayRepository.save(model);
        return this.gatewayConverter.toDTO(model);
    }

    @Override
    public GatewayDTO update(final String code, final GatewayDTO dto) {
        Gateway model = this.gatewayRepository.findByCode(code);
        if (null == model) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_404001, new Object[] {code });
        }
        dto.setId(model.getId());
        dto.setUpdateTime(new Date());
        model = this.gatewayConverter.toModel(dto);
        this.gatewayRepository.save(model);
        return this.gatewayConverter.toDTO(model);
    }

    @Override
    public void delete(final String code) {
        final Gateway model = this.gatewayRepository.findByCode(code);
        if (null == model) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_404001, new Object[] {code });
        }
        // 异步发送网关删除事件
        this.postDeleteEvent(model);
        model.getResources().stream().forEach(r -> {
            r.setGateway(null);
        });
        this.gatewayRepository.delete(model);
    }

    @Override
    public void enable(final String code) {
        final Gateway model = this.gatewayRepository.findByCode(code);
        if (null == model) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_404001, new Object[] {code });
        }
        model.setDisabled(false);
        model.setUpdateTime(new Date());
        this.gatewayRepository.save(model);
        // 异步发送状态改变事件
        this.postStatusChangedEvent(model);
    }

    @Override
    public void disable(final String code) {
        final Gateway model = this.gatewayRepository.findByCode(code);
        if (null == model) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_404001, new Object[] {code });
        }
        model.setDisabled(true);
        model.setUpdateTime(new Date());
        this.gatewayRepository.save(model);
        this.postStatusChangedEvent(model);
    }

    /**
     * 删除网关时，发送网关删除事件
     *
     * @param gateway
     */
    private void postDeleteEvent(final Gateway gateway) {
        final GatewayDeleteEvent event = new GatewayDeleteEvent();
        event.setGateway(gateway.getCode());
        gateway.getResources().stream().filter(it -> !it.isDisabled()).map(it -> it.getCode())
                .forEach(event.getResources()::add);
        this.template.convertAndSend(this.redisSubscribeChannel,
                JsonUtils.pojoToJson(new EventWrapper(EventType.gatewayDelete, JsonUtils.pojoToJson(event))));
        // this.modelUpdateEventBus.post(event);
    }

    /**
     * 状态改变时，发送网关状态变更事件
     *
     * @param gateway
     */
    private void postStatusChangedEvent(final Gateway gateway) {

        final GatewayStatusChangedEvent event = new GatewayStatusChangedEvent();
        event.setGateway(gateway.getCode());
        event.setDisabled(gateway.isDisabled());
        event.setBatched(gateway.isBatched());
        gateway.getResources().stream().filter(r -> !r.isDisabled()).forEach(r -> {
            event.getResources().put(r.getCode(), r.getExecutionCycle());
        });

        this.template.convertAndSend(this.redisSubscribeChannel,
                JsonUtils.pojoToJson(new EventWrapper(EventType.gatewayStatusChanged, JsonUtils.pojoToJson(event))));

        //this.modelUpdateEventBus.post(event);
    }

    /**
     * 状态批处理时，发送网关批处理变更事件
     *
     * @param gateway
     */
    private void postBatchChangedEvent(final Gateway gateway) {

        if (!gateway.isDisabled()) {
            final GatewayBatchChangedEvent event = new GatewayBatchChangedEvent();
            event.setGateway(gateway.getCode());
            event.setDisabled(gateway.isDisabled());
            event.setBatched(gateway.isBatched());
            gateway.getResources().stream().filter(r -> !r.isDisabled()).forEach(r -> {
                event.getResources().put(r.getCode(), r.getExecutionCycle());
            });
            //this.modelUpdateEventBus.post(event);
            this.template.convertAndSend(this.redisSubscribeChannel,
                    JsonUtils.pojoToJson(new EventWrapper(EventType.gatewayBatchChanged, JsonUtils.pojoToJson(event))));
        }
    }

    @Override
    public void importTpl(final MultipartFile excel) {
        final StringBuffer errmsg = new StringBuffer();
        final List<Gateway> listCache = Lists.newArrayList();
        try {
            try {
                ExcelPoiBoot.ImportBuilder(excel.getInputStream(), GatewayPortVO.class)
                        .importExcel(new ImportFunction<GatewayPortVO>() {
                            @Override
                            public void onProcess(final int sheetIndex, final int rowIndex,
                                    final GatewayPortVO portVO) {
                                final Gateway gateway = GatewayServiceImpl.this.gatewayConverter.toModel(portVO);
                                listCache.add(gateway);
                            }

                            /**
                             * @param errorEntity 错误信息实体
                             */
                            @Override
                            public void onError(final ErrorEntity errorEntity) {
                                GatewayServiceImpl.log.error("网关导入错误：{}", JsonUtils.pojoToJson(errorEntity));
                                errmsg.append(errorEntity.getErrorMessage()).append("\n");
                            }
                        });
            } catch (final IOException e) {
                throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_406112, new Object[] {e.getMessage() });
            }
            this.gatewayRepository.saveAll(listCache);
        } finally {
            listCache.clear();
        }
        if (StringUtils.isNotBlank(errmsg.toString())) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_406112, new Object[] {errmsg.toString() });
        }

    }

    @Override
    public SXSSFWorkbook getTpl() {
        final String sheetName = DatabusCoreConstant.TPL_NAME_GATEWAY;
        final Map<Integer, String[]> dropDowns = Maps.newHashMap();
        final String[] physicalConnTypes =
                Arrays.asList(PhysicalConnType.values()).stream().map(value -> value.toString().toLowerCase())
                        .toArray(String[]::new);

        dropDowns.put(1, physicalConnTypes);
        final String[] workModes =
                Arrays.asList(ResourceWorkMode.values()).stream().map(value -> value.toString().toLowerCase())
                        .toArray(String[]::new);

        dropDowns.put(3, workModes);

        final SXSSFWorkbook workbook =
                ExcelUtil.createExcelTemplate(sheetName, DatabusCoreConstant.GATEWAY_TPL_COLUMN, dropDowns);
        return workbook;
    }

    @Override
    public void export(final Searchable searchable, final Pageable pageable, final HttpServletResponse response) {
        final StringBuffer fileName = new StringBuffer("网关");
        fileName.append(DatabusCoreConstant.EXPORT_SPLITOR)
                .append(DateUtil.currentTimestamp2String(DateUtil.PATTERN_NULL_CONN));
        ExcelBoot
                .ExportBuilder(response, fileName.toString(), GatewayPortVO.class)
                .exportMultiSheetResponse(searchable, new ExportFunction<Searchable, Gateway>() {
                    @Override
                    public List<Gateway> pageQuery(final Searchable searchable, final int pageNum, final int pageSize) {
                        final QPageRequest qPageRequest = new QPageRequest(pageNum - 1, pageSize);
                        final Page<Gateway> models =
                                GatewayServiceImpl.this.gatewayRepository.search(qPageRequest, searchable);
                        return models.getContent();
                    }

                    @Override
                    public GatewayPortVO convert(final Gateway model) {
                        return GatewayServiceImpl.this.gatewayConverter.toVO(model);
                    }
                });

    }

    @Override
    public List<GatewayDTO> findAllEnabledAndPulled(int node) {
        final List<Gateway> models =
                this.gatewayRepository.findByWorkModeAndNodeAndDisabledIsFalse(ResourceWorkMode.PULL, node);
        return this.gatewayConverter.toListDTO(models);
    }

    @Override
    public void batch(String code, boolean batched) {
        final Gateway model = this.gatewayRepository.findByCode(code);
        if (null == model) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_404001, new Object[] {code });
        }
        model.setBatched(batched);
        model.setUpdateTime(new Date());
        this.gatewayRepository.save(model);
        // 异步发送状态改变事件
        this.postBatchChangedEvent(model);

    }

}
